package jxkjsync

import (
	"context"
	"fmt"
	"sort"
	"sync"
	"sync/atomic"
	"time"
)

type queueWorker struct {
	configSuffix           string
	mw                     WorkerWindow
	isRun                  bool
	isProcess              int32
	amBeginning            int32
	amEnding               int32
	amSaving               int32
	interval, times        int
	jobCtx                 context.Context
	jobCancel              context.CancelFunc
	endErr                 error
	jobQueueMap            map[*SyncQueue][]jobInfo
	jobLen                 int
	exceptionHandle        func(err error)
	beforeStart, afterStop func() error
	jobFinishHook          func() error
	logger                 Logger
	prefixSyncedCache      string
}

func (w *queueWorker) setLogger(logger Logger) {
	w.logger = logger
}
func (w *queueWorker) setPrefixSyncedCache(px string) {
	w.prefixSyncedCache = px
}
func (w *queueWorker) SetBeforeStart(f func() error)        { w.beforeStart = f }
func (w *queueWorker) SetAfterStop(f func() error)          { w.afterStop = f }
func (w *queueWorker) SetExceptionHandle(f func(err error)) { w.exceptionHandle = f }
func (w *queueWorker) SetJobFinishHook(f func() error)      { w.jobFinishHook = f }
func (w *queueWorker) HandleException(err error) {
	if w.exceptionHandle != nil {
		w.exceptionHandle(err)
	}
}
func (w *queueWorker) DoJobFinishHook() {
	if w.jobFinishHook != nil {
		w.jobFinishHook()
	}
}
func (w *queueWorker) DoAfterStop() {
	if w.afterStop != nil {
		w.afterStop()
	}
}
func (w *queueWorker) DoBeforeStart() error {
	if w.beforeStart != nil {
		return w.beforeStart()
	}
	return nil
}

func (w *queueWorker) IsRun() bool {
	return w.isRun
}
func (w *queueWorker) Error() error {
	return w.endErr
}
func (w *queueWorker) DoAwake(switchMap map[string]bool, sqlMap map[string]string) {
	if !w.isRun {
		w.DoStart(switchMap, sqlMap)
	}
}

func (w *queueWorker) DoStart(switchMap map[string]bool, sqlMap map[string]string) {
	go func() {
		defer fmt.Println("doStart end ")
		if !atomic.CompareAndSwapInt32(&w.amBeginning, 0, 1) {
			w.mw.NotifyInfo("同步开启中...")
			return
		}
		defer atomic.StoreInt32(&w.amBeginning, 0)
		isOk := false
		if w.isRun {
			w.mw.NotifyInfo("同步任务进行中")
			return
		}
		w.mw.ShowRunInfo("同步启动中..")
		w.mw.ShowReading()
		defer func() {
			if !isOk {
				//w.mw.ShowRunInfo("同步启动失败..")
				w.mw.ShowStopping()
			} else {
				w.isRun = true
				w.mw.ShowRunning()
			}
		}()
		w.mw.ShowRunInfo("同步任务初始化..")
		if err := w.DoBeforeStart(); err != nil {
			w.mw.NotifyInfo(err.Error())
			return
		}
		if len(switchMap) == 0 {
			w.mw.NotifyInfo("未开启同步项" + "(20009)")
			return
		}
		hasOpen := false
		for _, open := range switchMap {
			if open {
				hasOpen = open
			}
		}
		if !hasOpen {
			w.mw.NotifyInfo("未开启同步项" + "(20010)")
			return
		}

		//if lErr := initialize.LoadSqlConfig(); lErr != nil {
		//	w.logger.Error(lErr)
		//	w.mw.ShowRunInfo(lErr.Error())
		//	w.mw.NotifyInfo(lErr.Error())
		//	return
		//}

		jobQueueMap := make(map[*SyncQueue][]jobInfo)
		jobQueueExistMap := make(map[string]bool)
		jobLen := 0
		for _, t := range syncTypes {
			if open := switchMap[t]; open {
				srv := GetSyncService(t)
				if srv == nil {
					w.mw.NotifyInfo("暂不支持同步数据类型: " + t + "(20004)")
					return
				}
				if !srv.WantRun() {
					continue
				}
				if srv.GetQueue() == nil {
					w.mw.NotifyInfo("同步数据类型:" + srv.GetNameZh() + " 未设定队列信息" + "(20029)")
					return
				}
				if srv.GetStep() == 0 {
					w.mw.NotifyInfo("同步数据类型:" + srv.GetNameZh() + " 未设定步骤信息" + "(20027)")
					return
				}

				jb := jobInfo{srv: srv}
				if srv.WantSql() {
					if _, ok := sqlMap[t]; !ok {
						w.mw.NotifyInfo("同步数据类型: " + srv.GetNameZh() + " 未配置查询语句" + "(20000)")
						return
					} else {
						if err := srv.ValidateSql(sqlMap[t]); err != nil {
							w.mw.NotifyInfo("同步数据类型: " + srv.GetNameZh() + " " + err.Error())
							return
						}
						jb.sql = sqlMap[t]
					}
				}
				//附属同步语句
				if srv.GetSubType() != "" {
					if _, ok := sqlMap[srv.GetSubType()]; !ok {
						w.mw.NotifyInfo("同步数据类型:" + (GetSyncService(srv.GetSubType())).GetNameZh() + " 未配置查询语句" + "(20000)")
						return
					}
					jb.sql = jb.sql + "&&" + sqlMap[srv.GetSubType()]
				}

				if _, ok := jobQueueExistMap[srv.GetQueue().GetName()]; !ok {
					jobQueueMap[srv.GetQueue()] = []jobInfo{jb}
					jobQueueExistMap[srv.GetQueue().GetName()] = true
				} else {
					jobQueueMap[srv.GetQueue()] = append(jobQueueMap[srv.GetQueue()], jb)
				}

				jobLen++
			}
		}
		if jobLen == 0 {
			w.mw.NotifyInfo("未开启同步项" + "(20013)")
			return
		}

		for _, jobList := range jobQueueMap {
			sort.Slice(jobList, func(i, j int) bool {
				return jobList[i].srv.GetStep() < jobList[j].srv.GetStep()
			})
		}
		jobs := "will sync:"

		for queue, jobList := range jobQueueMap {
			tt := ""
			num := 0
			for _, job := range jobList {
				tt += job.srv.GetNameZh() + "|"
				num++
			}
			jobs += fmt.Sprintf("[queue:%s,nul:%d,type:%s]", queue.GetName(), num, tt)
		}

		fmt.Println(jobs)
		w.logger.Info(jobs)
		fmt.Println(">>>>>>> len jobList:", jobLen)

		w.jobCtx, w.jobCancel = context.WithCancel(context.Background())

		w.jobQueueMap = jobQueueMap
		w.jobLen = jobLen
		go w.doSync()
		isOk = true
	}()
}

func (w *queueWorker) DoStop(err error) {
	go func() {
		defer fmt.Println("doStop end ")
		if !w.isRun {
			w.mw.NotifyInfo("同步未启动")
			return
		}
		if !atomic.CompareAndSwapInt32(&w.amEnding, 0, 1) {
			w.mw.NotifyInfo("同步正在结束中...")
			return
		}

		defer atomic.StoreInt32(&w.amEnding, 0)
		infoErr := "同步"
		if err != nil {
			infoErr += "异常"
		}
		w.mw.ShowRunInfo(infoErr + "停止中...")
		w.endErr = err
		ok := false
		w.mw.ShowReading()
		defer func() {
			if !ok {
				defer w.mw.ShowRunInfo(infoErr + "停止失败...")
				w.mw.ShowRunning()
			} else {
				defer w.mw.ShowRunInfo(infoErr + "已停止...")
				w.isRun = false
				w.mw.ShowStopping()
			}
		}()
		if w.jobCancel != nil {
			fmt.Println("stop call cancel")
			w.jobCancel()
		}
		//延迟更新结束状态 等待进行中任务完成
		for {
			if w.isProcess == 1 {
				fmt.Println("同步进程结束中-等候两秒")
				//w.mw.ShowRunInfo("同步任务结束中... ")
				time.Sleep(time.Second * 2)
			} else {
				break
			}
		}
		w.DoAfterStop()
		//initialize.CloseDb(g.DbErp)
		//initialize.CloseDb(g.DbShop)
		//cache.Close()
		ok = true
	}()
}

func (w *queueWorker) doSync() {
	defer fmt.Println("doSync end ")
	//time.Sleep(time.Second * 2)
	errCh := make(chan error, w.jobLen*2)

	//defer close(errCh)

	w.doJob("开始", errCh)
	return
	jobCh := make(chan string, 1)
	defer close(jobCh)
	//go w.loopJob(jobCh, errCh)
	n := time.Now()
	// 年月日时分秒作为任务前缀
	jobPrefix := fmt.Sprintf("%04d%02d%02d%02d%02d%02d", n.Year(), n.Month(), n.Day(), n.Hour(), n.Minute(), n.Second())
	w.times = 1
	delayTimes := 0
	jobCh <- fmt.Sprintf("wk%s%d-%d", jobPrefix, w.times, delayTimes)
	timer := time.Tick(time.Duration(w.interval) * time.Second)

	for {
		select {
		case <-w.jobCtx.Done():
			fmt.Println("out   done  received")
			return
		case err := <-errCh:
			w.isRun = true
			w.DoStop(err)
			w.mw.ShowRunInfo(err.Error())
			w.mw.NotifyInfo(err.Error())
			return
		case <-timer:
			if len(jobCh) > 0 {
				delayTimes++
				fmt.Println("Current task in progress,continue")
			} else {
				w.times++
				jobCh <- fmt.Sprintf("wk%s%d-%d", jobPrefix, w.times, delayTimes)
			}
		}
	}
}

func (w *queueWorker) doJob(jobID string, errCh chan error) {
	//s := time.Now()
	atomic.StoreInt32(&w.isProcess, 1)
	defer func() {
		atomic.StoreInt32(&w.isProcess, 0)
		fmt.Println("doJob end ")
	}()
	errHappenCh := make(chan struct{}, w.jobLen*2)
	defer func() {
		//if len(errHappenCh) > 0 {
		//	isSuccess = 0
		//}
		//_ = shop_query.SyncMetrics.Create(&shop_entity.SyncMetrics{
		//	JobID:       jobID,
		//	SyncService: "all-service",
		//	UsedSecond:  time.Since(s).Milliseconds(),
		//	StartTime:   s,
		//	EndTime:     time.Now(),
		//	IsSuccess:   int32(isSuccess),
		//	ErrorMsg:    "",
		//})
		close(errHappenCh)
	}()
	//pCtx, pCancel := context.WithCancel(context.TODO())
	select {

	case <-w.jobCtx.Done():
		w.logger.Info("doJob out  done  received")
		fmt.Println("doJob out  done  received")
		return
	default:
		w.mw.ShowRunInfo("同步 " + jobID)
		wg := sync.WaitGroup{}
		for que, jobs := range w.jobQueueMap {
			wg.Add(1)
			go func(queue *SyncQueue, jobList []jobInfo) {
				defer wg.Done()
				for {
					select {
					case err := <-errCh:
						w.isRun = true
						w.DoStop(err)
						w.mw.ShowRunInfo(err.Error())
						w.mw.NotifyInfo(err.Error())
						return
					case <-w.jobCtx.Done():
						w.logger.Info("doJob inner  done  received")
						fmt.Println("doJob inner  done  received")
						return
					//case <-pCtx.Done():
					//	w.logger.Info("doJob inner  pCtx done  received")
					//	fmt.Println("doJob inner  pCtx done  received")
					//	return
					default:
						queueSince := time.Now()
						fmt.Println("----start queue:", queue.GetName())
						for _, job := range jobList {
							//select {
							//case <-pCtx.Done():
							//	return
							//default:
							since := time.Now()

							syncErr := job.srv.SyncFunc(w.jobCtx, w.prefixSyncedCache+job.srv.GetName(), job.sql)
							//metric := &shop_entity.SyncMetrics{
							//	JobID:       id,
							//	SyncService: srv.GetName(),
							//	UsedSecond:  time.Since(since).Milliseconds(),
							//	StartTime:   since,
							//	EndTime:     time.Now(),
							//	IsSuccess:   1,
							//	ErrorMsg:    "",
							//}
							//if syncErr != nil {
							//	metric.IsSuccess = 0
							//	metric.ErrorMsg = syncErr.Error()
							//}
							//_ = shop_query.SyncMetrics.Create(metric)
							if syncErr != nil {

								w.logger.Error("dojob " + job.srv.GetName() + "  err:" + syncErr.Error())
								//errHappenCh <- struct{}{}
								//errCh <- errors.New("同步: " + job.srv.GetNameZh() + " 数据异常")
								w.mw.ShowRunInfo(job.srv.GetNameZh() + " :同步异常!")
								w.HandleException(fmt.Errorf("%s 同步异常:%s", job.srv.GetNameZh(), syncErr.Error()))
								//pCancel()
							}
							w.logger.Info(job.srv.GetName(), " alone used second:", time.Since(since).Seconds())
							//}
						}
						w.logger.Info("one queue :", queue.GetName(), " used second:", time.Since(queueSince).Seconds())
						fmt.Printf("one queue :%s used second:%f\n", queue.GetName(), time.Since(queueSince).Seconds())
						if queue.finishHook != nil {
							queue.finishHook()
						}
					}
					time.Sleep(queue.GetInterval())
				}
			}(que, jobs)
		}
		wg.Wait()
		//fmt.Printf("all queue used second:%f\n", time.Since(allSince).Seconds())
		//else {
		//	err := errors.New("暂不支持同步类型: " + job.dataType + "(20002)")
		//	fmt.Println(err)
		//	w.logger.Error(err)
		//	errCh <- err
		//	return
		//}

		//if len(errHappenCh) > 0 {
		//	w.mw.ShowRunInfo(fmt.Sprintf("任务异常中断,step:%+v", step))
		//	return
		//}
		//w.mw.ShowRunInfo(fmt.Sprintf("同步:%s,step:%+v", jobID, step))
	}
	//fmt.Println("syncData wait end order:", step)
	//w.logger.Info(fmt.Sprintf("end job order:%d ,id:%s ", step, jobID))
	//time.Sleep(time.Second * 2)
	//w.mw.ShowRunInfo("完成同步 " + jobID)
	//w.logger.Info(jobID, " all used second:", time.Since(s).Seconds())
	//{
	// 更新同步时间
	w.DoJobFinishHook()
	//_, _ = shop_query.Config.
	//	Where(shop_query.Config.ConfigKey.Eq("API_CONFIG")).
	//	Update(shop_query.Config.ModifyTime, time.Now().Unix())
	//}
}
